package com.audaque.springboot.foshanupload.web.flinkdemo.main;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.Comparator;
import java.util.Map;
import java.util.TreeMap;

public class FlinkWindowT1 {

    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//        // todo 获取kafka的配置属性
//        args = new String[]{"--input-topic", "dianyou_wxgz_test2", "--output-topic", "dianyou_wxgz_test3", "--bootstrap.servers", "172.10.4.141:9092,172.10.4.142:9092,172.10.4.143:9092",
//                "--zookeeper.connect", "172.10.4.63:2181,172.10.4.64:2181,172.10.4.65:2181", "--group.id", "cc"};
//
//        ParameterTool parameterTool = ParameterTool.fromArgs(args);
//        if (parameterTool.getNumberOfParameters() < 5) {
//            System.out.println("Missing parameters!" +
//                    "Usage: Kafka --input-topic <topic> --output-topic <topic> " +
//                    "--bootstrap.servers <kafka brokers> " +
//                    "--zookeeper.connect <zk quorum> --group.id <some id>");
//            return;
//        }
//        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


        // todo 指定输入数据为kafka topic
        DataStreamSource<String> inputStream = env.socketTextStream("localhost", 9999);

        DataStream<Tuple2<String, Integer>> ds = inputStream
                .flatMap(new LineSplitter()); //将输入语句split成一个一个单词并初始化count值为1的Tuple2<String, Integer>类型


        DataStream<Tuple2<String, Integer>> wcount = ds
                .keyBy(0)
                .window(SlidingProcessingTimeWindows.of(Time.seconds(6), Time.seconds(3)))
                //key之后的元素进入一个总时间长度为600s,每5s向后滑动一次的滑动窗口
                .sum(1);// 将相同的key的元素第二个count值相加


        wcount.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(1)))
                //所有key元素进入一个5s长的窗口（选5秒是因为上游窗口每5s计算一轮数据，topN窗口一次计算只统计一个窗口时间内的变化）
                .process(new TopNAllFunction(3))
                .print();
        //redis sink redis -> 接口

        env.execute();
    }//


    private static final class LineSplitter implements
            FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {

            out.collect(new Tuple2<String, Integer>(value, 1));
        }
    }

    private static class TopNAllFunction extends ProcessAllWindowFunction<Tuple2<String, Integer>, String, TimeWindow> {

        private int topSize = 3;

        public TopNAllFunction(int topSize) {

            this.topSize = topSize;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
        }

        @Override
        public void process(
                ProcessAllWindowFunction<Tuple2<String, Integer>, String, TimeWindow>.Context arg0,
                Iterable<Tuple2<String, Integer>> input,
                Collector<String> out) throws Exception {

            //treemap按照key降序排列，相同count值不覆盖
            TreeMap<Integer, Tuple2<String, Integer>> treemap = new TreeMap<Integer, Tuple2<String, Integer>>(
                    new Comparator<Integer>() {

                        @Override
                        public int compare(Integer y, Integer x) {
                            return (x < y) ? -1 : 1;
                        }

                    });

            //只保留前面TopN个元素

            for (Tuple2<String, Integer> element : input) {
                treemap.put(element.f1, element);
                if (treemap.size() > topSize) {
                    treemap.pollLastEntry();
                }
            }


            for (Map.Entry<Integer, Tuple2<String, Integer>> entry : treemap
                    .entrySet()) {
                out.collect("=================\n热销图书列表:\n" + new Timestamp(System.currentTimeMillis()) + treemap.toString() + "\n===============\n");
            }

        }


    }
}
